#include <ngx_config.h>
#include <ngx_core.h>
#include <assert.h>
#include <stdio.h>
#include <stddef.h>
#include <string.h>
#include <unistd.h>
#include <netinet/in.h>
#include <errno.h>
#include <sys/socket.h>
#include <netinet/ip.h>
#include <netinet/udp.h>
#include <netinet/tcp.h>
#include <netinet/if_ether.h>
#include <arpa/inet.h>
#include <pthread.h>
#include <syslog.h>
#include <sys/time.h>
#include <arpa/inet.h>
#include <errno.h>
#include <stdlib.h>
#include <ctype.h>
#include <pthread.h>
#include "worker.h"

//todo: try add conn_t* into worker, test the performace
struct _conn_t;
typedef struct _conn_t conn_t;

typedef uint16_t str_id_t;

struct _conn_t {
    conn_t     *next; 
    
    uint32_t    svr_addr;
    uint32_t    clt_addr;
    
    uint32_t    hash;
    uint16_t    clt_port;
    uint16_t    svr_port;
    
    size_t      vol_up;
    size_t      vol_dn;
    
    uint32_t    start_sec;
    uint16_t    start_msec;
    uint16_t    response_code;
    
    //ngx_queue_t timer_que; //?
    uint32_t    active_time;
    uint16_t    used;

    str_id_t    uri;
    str_id_t    host;
    str_id_t    referer;
    str_id_t    cache_control;
    str_id_t    cookie;
    str_id_t    content_length;
    str_id_t    content_type;
    str_id_t    user_agent;
};

typedef struct 
{
    pthread_spinlock_t lock;
    int                conn_cnt;
    conn_t            *conns;
} conn_tbl_item_t;


#define FROM_CLT  0
#define FROM_SVR  1
#define PAGE_SIZE (1024*2)
#define PAGE_NUM  (1024*1024)
#define worker_hashsize 500000

#define addr_hash(saddr, sport, daddr, dport) ((saddr ^ sport) ^ (daddr ^ dport))

#define ngx_str4cmp(m, c0, c1, c2, c3)                                        \
    m[0] == c0 && m[1] == c1 && m[2] == c2 && m[3] == c3

#define ngx_str5cmp(m, c0, c1, c2, c3, c4)                                    \
    m[0] == c0 && m[1] == c1 && m[2] == c2 && m[3] == c3 && m[4] == c4

uint32_t         current_sec;
uint32_t         timeout_requests;

conn_tbl_item_t *conn_tbl;
static void     *pages;

ngx_atomic_t     conn_num;
static pthread_spinlock_t page_lock;

unsigned  workers_cnt;
int       max_que_len;

worker_t  workers[64];


static void finish_request(worker_t *w, conn_t *c);
void worker_free_conn(conn_tbl_item_t *ti, conn_t *c);

static inline int addr_cmp(conn_t *c, struct iphdr* iph, struct tcphdr* tcph)
{
    return ((c->svr_addr == iph->saddr && c->svr_port == tcph->source
        && c->clt_addr == iph->daddr && c->clt_port == tcph->dest)
        || (c->svr_addr == iph->daddr && c->svr_port == tcph->dest
        && c->clt_addr == iph->saddr && c->clt_port == tcph->source));
}

void
ngx_log_error_core(ngx_uint_t level, ngx_log_t *log, ngx_err_t err,
    const char *fmt, ...)
{
    //dummy
}

static inline uint16_t
xgn_atoi(char *line, size_t n)
{
    ngx_int_t  value;

    if (n == 0) {
        return NGX_ERROR;
    }

    for (value = 0; n--; line++) {
        if (*line < '0' || *line > '9') {
            return NGX_ERROR;
        }

        value = value * 10 + (*line - '0');
    }

    if (value < 0) {
        return NGX_ERROR;

    } else {
        return value;
    }
}

void *worker_timeout_thread(void *p)
{
    conn_t *c;
    conn_t *c_next;
    int i = 0, j;
    
    while (1)
    {
        current_sec = (uint32_t)time(NULL);

        j = i + 500;
        if (j > worker_hashsize) j = worker_hashsize;
        
        for (; i < j; i++)
        {
            if (!conn_tbl[i].conns) continue;
            pthread_spin_lock(&conn_tbl[i].lock);
            for (c = conn_tbl[i].conns; c; c = c_next)
            {
                c_next = c->next;
                //if (c->active_time != 0)
                //{
                //    printf("c->active_time %d\n", current_sec - c->active_time);
                //}
                //else printf("hash no %d   que len %d\n", i, (int)conn_tbl[i].conn_cnt);
                if (current_sec - c->active_time >= 60)
                {
                    finish_request(&workers[0], c);
                    worker_free_conn(&conn_tbl[i], c);
                    timeout_requests++;
                }
            }
            pthread_spin_unlock(&conn_tbl[i].lock);
        }

        if (i == worker_hashsize) i = 0;
        
        sleep(1);
    }

    return NULL;
}

void *worker_print_thread(void *p)
{
    unsigned    i;
    worker_t   *w;
    worker_t    acc;
    size_t      last_total_bytes = 0;
    
    while (1)
    {
        memset(&acc, 0, sizeof(acc));
        for (i = 0; i < workers_cnt; i++)
        {
            w = &workers[i];
            acc.total_bytes += w->total_bytes;
            acc.drop_unestablished += w->drop_unestablished;
            acc.no_mem += w->no_mem;
            acc.req += w->req;
            acc.req_incomplete += w->req_incomplete;
        }
        printf("--------------------------------------\n");
        if (last_total_bytes == 0)
        {
            printf("speed               ?\n");
        }
        else
        {
            printf("speed               %ld Mbps\n", (acc.total_bytes - last_total_bytes) / 1024 / 1024 * 8);
        }
        last_total_bytes = acc.total_bytes;
        printf("total_bytes         %ld\n", acc.total_bytes);
        printf("drop_unestablished  %d packets\n", acc.drop_unestablished);
        printf("no_mem              %d\n", acc.no_mem);
        printf("connections         %d\n", (int)conn_num);
        printf("requests            %d\n", acc.req);
        printf("req_incomplete      %d\n", acc.req_incomplete);
        printf("max_que_len         %d\n", max_que_len);
        printf("timeout_requests    %d\n", timeout_requests);
        sleep(1);
    }

    return NULL;
}

void worker_global_init()
{
    int       i;
	pthread_t thr;
    void      *p;
    
    pthread_spin_init(&page_lock, 0);
    for (i = 0; i < PAGE_NUM; i++)
    {
        p = malloc(PAGE_SIZE);
        if (!p)
        {
            printf("Not enough memory\n");
            exit(1);
        }
        *(void**)p = pages;
        pages = p;
    }
    
    conn_tbl = (conn_tbl_item_t*)malloc(sizeof(conn_tbl_item_t) * worker_hashsize);
    if (!conn_tbl)
    {
        printf("fail to create connection table.\n");
        exit(1);
    }

    for (i = 0; i < worker_hashsize; i++)
    {
        pthread_spin_init(&conn_tbl[i].lock, 0);
        conn_tbl[i].conns = NULL;
    }

	pthread_create(&thr, NULL, worker_print_thread, NULL);
	pthread_create(&thr, NULL, worker_timeout_thread, NULL);
}

worker_t *worker_create(const char *ifname, int thrid)
{
    worker_t *w = NULL;
    char      path[100];

    if (workers_cnt >= sizeof(workers)/sizeof(workers[0]))
    {
        fprintf(stderr, "No more workers\n");
        return NULL;
    }
    
    w = &workers[workers_cnt++];
    memset(w, 0, sizeof(worker_t));

    sprintf(path, "httplog_%s_%d.txt", ifname, thrid);
    w->fp = fopen(path, "a");
    if (!w->fp)
    {
        fprintf(stderr, "Failed to open %s for write\n", path);
        return NULL;
    }

    return w;
}

static inline void *worker_get_page()
{
    void *p;
    
    pthread_spin_lock(&page_lock);
    
    if (pages)
    {
        p = pages;
        pages = *(void**)pages;
    }
    else
    {
        p = NULL;
    }
    
    pthread_spin_unlock(&page_lock);
    return p;
}

static inline void worker_free_page(void *page)
{
    pthread_spin_lock(&page_lock);

    *(void**)page = pages;
    pages = page;
    
    pthread_spin_unlock(&page_lock);
}

static str_id_t worker_strdup(conn_t *c, char *data, size_t len)
{
    char *p;
    str_id_t sid;
    
    if (PAGE_SIZE - c->used >= (int)len + 1)
    {
        p = (char*)c + c->used;
        memcpy(p, data, len);
        p[len] = '\0';
        sid = c->used;
        c->used += len + 1;
        return sid;
    }
    else
    {
        return 0;
    }
}

static inline void worker_reset_conn(conn_t *c)
{
    c->vol_up = 0;
    c->vol_dn = 0;
    c->used = sizeof(conn_t);
    c->uri = NULL;
    c->host = NULL;
    c->referer = NULL;
    c->cache_control = NULL;
    c->cookie = NULL;
    c->content_length = NULL;
    c->content_type = NULL;
    c->user_agent = NULL;
    c->response_code = 0;
}

void worker_free_conn(conn_tbl_item_t *ti, conn_t *c)
{
    conn_t **pp;

    for (pp = &ti->conns; *pp != c && *pp != NULL; pp = (conn_t **)*pp)
    {}
    if (*pp == NULL)
    {
        return;
    }
    
    *pp = c->next;

    worker_free_page(c);
    ngx_atomic_fetch_add(&conn_num, -1);
}

static inline void worker_print_str(FILE *fp, conn_t *c, const str_id_t sid)
{
    if (sid)
    {
        fprintf(fp, "%s\001", (char*)c + sid);
    }
    else
    {
        fprintf(fp, "N/A\001");
    }
}

static void finish_request(worker_t *w, conn_t *c)
{
    struct timeval tv;
    char SrcIP[16];
    char DstIP[16];
    time_t t;
    struct tm st;

    if (c->response_code == 0)
    {
        w->req_incomplete++;
    }

    if (c->host && c->uri)
    {
        gettimeofday(&tv, NULL);
        w->req++;
      /*

    sznodeid + "\001"
    + SrcIP + "\001" + DstIP + "\001"
    + SrcPort + "\001" + DstPort + "\001"
    + szTranPro +  "\001"  + szAppPro +  "\001"  
    + sdf.format(ltimer) + "\001" + vistTime +  "\001"  
    + Upsteam + "\001" + downstream + "\001" 
    + Host + "\001"  + User-agent + "\001" 
    + szRspStatus + "\001"
    + Url + "\001"
    + Refer + "\001"
    + Refer + "\001"
    + szCacheControl + "\001" 
    +  szContentLength + "\001" +  szContentType+ "\001" 
    + "\001"+ "\001"+ "\001";

    */
        inet_ntop(AF_INET, &c->clt_addr, SrcIP, sizeof(SrcIP));
        inet_ntop(AF_INET, &c->svr_addr, DstIP, sizeof(DstIP));

        t = c->start_sec;
        localtime_r(&t, &st);
        
        fprintf(w->fp, 
            "1\001%s\001%s\001%d\001%d\001TCP\001HTTP\001"
            "%04d%02d%02d%02d%02d%02d\001" //sdf.format(ltimer)
            "%d\001" //vistTime
            "%ld\001%ld\001" //upstream downstream
            ,
            SrcIP, DstIP, ntohs(c->clt_port), ntohs(c->svr_port),
            st.tm_year + 1900, st.tm_mon + 1, st.tm_mday, st.tm_hour, st.tm_min, st.tm_sec,
            (int)(tv.tv_sec - c->start_sec) * 1000 + (int)(tv.tv_usec / 1000) - c->start_msec,
            (long)c->vol_up, (long)c->vol_dn);
        worker_print_str(w->fp, c, c->host);
        worker_print_str(w->fp, c, c->user_agent);
        if (c->response_code)
            fprintf(w->fp, "%d\001", c->response_code);
        else
            fprintf(w->fp, "N/A\001");
        worker_print_str(w->fp, c, c->uri);
        worker_print_str(w->fp, c, c->referer);
        worker_print_str(w->fp, c, c->referer); //dup intended
        worker_print_str(w->fp, c, c->cache_control);
        worker_print_str(w->fp, c, c->content_length);
        worker_print_str(w->fp, c, c->content_type);
        fprintf(w->fp, "\001\001\001\n");
    }
    
}

static void process_clt_pkt(worker_t *w, conn_t *c, char *p, size_t len)
{
    char    *end = p + len;
    char    *t;
    struct timeval tv;
    str_id_t *ps;
    
    gettimeofday(&tv, NULL);
    c->start_sec = (uint32_t)tv.tv_sec;
    c->start_msec = tv.tv_usec / 1000;

    while (p < end && *p != ' ') p++; //skip GET
    while (p < end && *p == ' ') p++; //skip spaces after GET
    t = p;
    while (p < end && *p != ' ') p++;
    if (p < end)
        c->uri = worker_strdup(c, t, p - t);
    while (p < end && *p != '\n') p++;
    p++; //request line ends

    while (p < end)
    {
        t = p;
        while (p < end && *p != ':') p++;
        if (p >= end) return;
        
        //to optimize strncasecmp
        if (p - t == sizeof("host") - 1 
            && strncasecmp(t, "host", sizeof("host") - 1) == 0)
        {
            ps = &c->host;
        }
        else if (p - t == sizeof("cookie") - 1 
            && strncasecmp(t, "cookie", sizeof("cookie") - 1) == 0)
        {
            ps = &c->cookie;
        }
        else if (p - t == sizeof("referer") - 1 
            && strncasecmp(t, "referer", sizeof("referer") - 1) == 0)
        {
            ps = &c->referer;
        }
        else if (p - t == sizeof("user-agent") - 1  
            && strncasecmp(t, "user-agent", sizeof("user-agent") - 1) == 0)
        {
            ps = &c->user_agent;
        }
        else
        {
            ps = NULL;
        }
        
        p++;
        if (*p != ' ') 
            t = p;
        else
            t = p + 1;
        while (p < end && *p != '\r') p++; //to the end of line
        if (p >= end) return;
        if (ps)
        {
            *ps = worker_strdup(c, t, p - t);
        }
        p+=2; //skip \r\n
    }
}

static void process_svr_pkt(worker_t *w, conn_t *c, char *p, size_t len)
{
    char    *end = p + len;
    char    *t;
    str_id_t *ps;

    p += sizeof("HTTP/1.0") - 1;
    while (p < end && *p == ' ') p++;
    t = p;
    while (p < end && *p != ' ') p++;
    
    if (p < end)
        c->response_code = xgn_atoi(t, p - t);

    while (p < end && *p != '\n') p++;
    p++; //response line ends

    while (p < end)
    {
        
        t = p;
        while (p < end && *p != ':') p++;
        if (p >= end) return;
        
        if (p - t == sizeof("content-type") - 1
            && strncasecmp(t, "content-type", sizeof("content-type") - 1) == 0)
        {
            ps = &c->content_type;
        }
        else if (p - t == sizeof("cache-control") - 1 
            && strncasecmp(t, "cache-control", sizeof("cache-control") - 1) == 0)
        {
            ps = &c->cache_control;
        }
        else if (p - t == sizeof("content-length") - 1 
            && strncasecmp(t, "content-length", sizeof("content-length") - 1) == 0)
        {
            ps = &c->content_length;
        }
        else
        {
            ps = NULL;
        }
        
        p++;
        if (*p != ' ') 
            t = p;
        else
            t = p + 1;
        while (p < end && *p != '\r') p++; //to the end of line
        if (p >= end) return;
        if (ps)
        {
            *ps = worker_strdup(c, t, p - t);
        }
        p+=2; //skip \r\n
    }
}

void worker_process_pkt(worker_t *w, struct iphdr *iph, struct tcphdr *tcph, u_char *data, size_t len)
{
    conn_t             *c;
    conn_tbl_item_t    *ti;
    int                 from;
    uint32_t            hash;
    int                 ip_tot_len = ntohs(iph->tot_len);
    
    w->total_bytes += sizeof(struct ethhdr) + ip_tot_len;

    if (len == 0 && !tcph->fin && !tcph->rst)
    {
        return;
    }

    hash = addr_hash(iph->saddr, tcph->source, iph->daddr, tcph->dest) % worker_hashsize;
    ti = &conn_tbl[hash];
    pthread_spin_lock(&ti->lock);

    for (c = ti->conns; c != NULL; c = (conn_t*)c->next)
    {
        if (iph->saddr == c->svr_addr && tcph->source == c->svr_port
            && iph->daddr == c->clt_addr && tcph->dest == c->clt_port)
        {
            from = FROM_SVR;
            c->vol_dn += ip_tot_len;
            break;
        }
        else if (iph->saddr == c->clt_addr && tcph->source == c->clt_port
            && iph->daddr == c->svr_addr && tcph->dest == c->svr_port)
        {
            if ((len >= 4 && ngx_str4cmp(data, 'G', 'E', 'T', ' '))
                || (len >= 5 && ngx_str5cmp(data, 'P', 'O', 'S', 'T', ' '))
                || (len >= 5 && ngx_str5cmp(data, 'H', 'E', 'A', 'D', ' ')))
            {
                finish_request(w, c);
                worker_reset_conn(c);    
            }
            from = FROM_CLT;
            c->vol_up += ip_tot_len;
            break;
        }
    }

    if (!c)
    {
        //not found
        if ((len >= 4 && ngx_str4cmp(data, 'G', 'E', 'T', ' '))
            || (len >= 5 && ngx_str5cmp(data, 'P', 'O', 'S', 'T', ' '))
            || (len >= 5 && ngx_str5cmp(data, 'H', 'E', 'A', 'D', ' ')))
        {
            c = (conn_t*)worker_get_page();
            if (!c)
            {
                w->no_mem++;
                pthread_spin_unlock(&ti->lock);
                return;
            }
            worker_reset_conn(c);
            
            c->next = ti->conns;
            c->hash = hash;
            c->clt_addr = iph->saddr;
            c->clt_port = tcph->source;
            c->svr_addr = iph->daddr;
            c->svr_port = tcph->dest;

            ti->conns = c;
            ti->conn_cnt++;
            ngx_atomic_fetch_add(&conn_num, 1);
            if (ti->conn_cnt > max_que_len)
            {
                max_que_len = ti->conn_cnt;
            }
            from = FROM_CLT;
        }
        else
        {
            w->drop_unestablished++;
            pthread_spin_unlock(&ti->lock);
            return;
        }
    }

    c->active_time = current_sec;
    
    if (len == 0)
    {
        //if (tcph->fin || tcph->rst)
        {
            finish_request(w, c);
            worker_free_conn(ti, c);
        }
    }
    else if (from == FROM_CLT && c->uri == NULL)
    {
        process_clt_pkt(w, c, (char*)data, len);
    }
    else if (from == FROM_SVR && c->response_code == 0)
    {
        process_svr_pkt(w, c, (char*)data, len);
    }

    pthread_spin_unlock(&ti->lock);
    
    return;
}

